package com.springboot.demo.io.socketmultiplex;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多路复用器组
 *
 * @author supu
 * @since 2020-11-13 13:34
 **/
public class SelectorThreadGroup {
    /**
     * 事务id，用于计算轮训hash值
     */
    private final AtomicInteger xid = new AtomicInteger(0);

    /**
     * 多路复用器工作组
     */
    private SelectorThreadGroup workGroup = this;

    /**
     * 该工作组拥有的 selector
     */
    private SelectorThread[] selectorThreadList;

    public SelectorThreadGroup(int selectorNum) {
        selectorThreadList = new SelectorThread[selectorNum];
        for (int i = 0; i < selectorNum; i++) {
            selectorThreadList[i] = new SelectorThread(this);
            // 启动所有的 selector
            new Thread(selectorThreadList[i]).start();
        }
    }

    public void setWorkGroup(SelectorThreadGroup workGroup) {
        this.workGroup = workGroup;
    }

    /**
     * socket服务绑定端口
     * @param port
     */
    public void bind(int port){
        try {
            ServerSocketChannel socketChannel = ServerSocketChannel.open();
            socketChannel.configureBlocking(false);
            socketChannel.bind(new InetSocketAddress(port));

            // 服务注册到 selector 上
            registerSelector(socketChannel);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void registerSelector(Channel channel) {
        SelectorThread selectorThread;

        if(channel instanceof ServerSocketChannel){
            // boss selector 注册
            selectorThread = selectBossSelectorGroup();
        } else {
            // 普通工作组 selector 注册
            selectorThread = selectSelectorGroup();
        }

        selectorThread.channelQueue.offer(channel);

        // 唤醒阻塞的 selector
        selectorThread.selector.wakeup();
    }

    private SelectorThread selectSelectorGroup() {
        int num = xid.getAndIncrement() % workGroup.selectorThreadList.length;
        return workGroup.selectorThreadList[num];
    }

    private SelectorThread selectBossSelectorGroup() {
        int num = xid.getAndIncrement() % selectorThreadList.length;
        return selectorThreadList[num];
    }
}
